<?php
/*
	本插件：$this->model('websocket')->方法名();
	其他地方调用：model('jy_agentmcp/websocket')->方法名();
*/
namespace jy_agentmcp;
defined('BY_JYA') or exit('error');
class websocket extends \plugin{
	private $socket;
    private $host;
    private $port;
    private $path;
    private $ssl;
    private $connected = false; // 连接状态标识
    private $token; // 鉴权令牌

    public function __construct($url, $token = '') {
        $parsedUrl = parse_url($url);
        if (!$parsedUrl || !isset($parsedUrl['host'])) {
            throw new \InvalidArgumentException("无效的WebSocket URL: {$url}");
        }
        $this->host = $parsedUrl['host'];
        $this->port = isset($parsedUrl['port']) ? $parsedUrl['port'] : 80;
        $this->path = isset($parsedUrl['path']) ? $parsedUrl['path'] : '/';
        $this->ssl = ($parsedUrl['scheme'] ?? '') === 'wss';  
        $this->token = $token;

        // 处理查询参数
        if (!empty($parsedUrl['query'])) {
            $this->path .= '?' . $parsedUrl['query'];
        }
        
        // wss默认端口处理
        if ($this->ssl && $this->port == 80) {
            $this->port = 443;
        }
        
        $this->socket = null;
    }

    /**
     * 建立WebSocket连接
     * @return bool 连接是否成功
     */
    public function connect() {
        $this->close(); // 确保关闭之前的连接

        $transport = $this->ssl ? 'ssl://' : '';
        $this->socket = fsockopen(
            $transport . $this->host,
            $this->port,
            $errno,
            $errstr,
            10 // 连接超时时间
        );
        
        if (!$this->socket) {
            $this->logError("连接失败: [{$errno}] {$errstr}");
            return false;
        }
        
        // 设置流超时（读取/写入超时）
        stream_set_timeout($this->socket, 30);
        stream_set_blocking($this->socket, 0); // 非阻塞模式

        // 发送握手请求
        $key = base64_encode(openssl_random_pseudo_bytes(16));
        $headers = [
            "GET {$this->path} HTTP/1.1",
            "Host: {$this->host}",
            "Upgrade: websocket",
            "Connection: Upgrade",
            "Sec-WebSocket-Key: {$key}",
            "Sec-WebSocket-Version: 13",
            "Origin: http://{$this->host}"
        ];
        
        // 添加鉴权头
        if (!empty($this->token)) {
            $headers[] = "Authorization: Bearer {$this->token}";
        }
        
        $handshake = implode("\r\n", $headers) . "\r\n\r\n";
        if (fwrite($this->socket, $handshake) !== strlen($handshake)) {
            $this->logError("握手请求发送失败");
            $this->close();
            return false;
        }

        // 读取握手响应（带超时控制）
        $response = '';
        $timeout = 5; // 握手超时时间(秒)
        $startTime = time();
        do {
            $buffer = fread($this->socket, 1024);
            if ($buffer !== false) {
                $response .= $buffer;
            }
            // 检查是否读取到完整响应或超时
            if (strpos($response, "\r\n\r\n") !== false || time() - $startTime > $timeout) {
                break;
            }
            usleep(100000); // 100ms间隔重试
        } while (true);

        // 验证握手结果
        if (strpos($response, ' 101 ') === false || strpos($response, 'Sec-WebSocket-Accept: ') === false) {
            $this->logError("握手失败，响应: " . substr($response, 0, 200));
            $this->close();
            return false;
        }

        $this->connected = true;
        $this->logInfo("握手成功");
        return true;
    }

    /**
     * 发送数据
     * @param string $data 要发送的数据
     * @return bool 是否发送成功
     */
    public function send($data) {
        if (!$this->isConnected()) {
            $this->logError("发送失败：未建立连接");
            return false;
        }
        
        $payload = $this->encode($data);
        $bytesWritten = fwrite($this->socket, $payload);
        if ($bytesWritten !== strlen($payload)) {
            $this->logError("数据发送不完整，发送了{$bytesWritten}/" . strlen($payload) . "字节");
            $this->connected = false; // 标记连接异常
            return false;
        }
        return true;
    }

    /**
     * 接收数据
     * @param int $timeout 超时时间(秒)
     * @return string|false 接收的数据或false(连接异常)
     */
    public function receive($timeout = 30) {
        if (!$this->isConnected()) {
            return false;
        }

        // 检查流状态
        $status = stream_get_meta_data($this->socket);
        if ($status['timed_out'] || $status['eof']) {
            $this->logError("接收超时或连接已关闭");
            $this->connected = false;
            return false;
        }

        // 读取头部（2字节）
        $header = $this->readExactly(2);
        if ($header === false) {
            return false;
        }
        
        $byte1 = ord($header[0]);
        $byte2 = ord($header[1]);

        $fin = ($byte1 & 0x80) != 0;
        $masked = ($byte2 & 0x80) != 0;
        $payloadLength = $byte2 & 0x7F;

        // 处理扩展长度
        if ($payloadLength == 126) {
            $extend = $this->readExactly(2);
            if ($extend === false) return false;
            $payloadLength = unpack('n', $extend)[1];
        } elseif ($payloadLength == 127) {
            $extend = $this->readExactly(8);
            if ($extend === false) return false;
            $payloadLength = unpack('J', $extend)[1];
        }
        
        // 读取掩码
        $mask = '';
        if ($masked) {
            $mask = $this->readExactly(4);
            if ($mask === false) return false;
        }

        // 读取payload
        $payload = $this->readExactly($payloadLength);
        if ($payload === false) return false;

        // 解码
        if ($masked) {
            $payload = $this->applyMask($payload, $mask);
        }
        // 新增：检测编码并转换为UTF-8
        $encoding = mb_detect_encoding($payload, ['UTF-8', 'ISO-8859-1', 'Windows-1252']);
        if ($encoding && $encoding !== 'UTF-8') {
            $payload = mb_convert_encoding($payload, 'UTF-8', $encoding);
        }

        return $payload;
    }

    /**
     * 处理掩码解码
     */
    private function applyMask($payload, $mask) {
        $payloadLength = strlen($payload);
        $result = '';
        
        for ($i = 0; $i < $payloadLength; $i++) {
            $maskIndex = $i % 4;
            $maskByte = ord($mask[$maskIndex]);
            $payloadByte = ord($payload[$i]);
            $result .= chr($payloadByte ^ $maskByte);
        }
        
        return $result;
    }
    /**
     * 精确读取指定长度的数据
     * @param int $length 要读取的长度
     * @return string|false 读取的数据或false
     */
    private function readExactly($length) {
        if ($length <= 0) return '';

        $data = '';
        $startTime = time();
        while (strlen($data) < $length) {
            $need = $length - strlen($data);
            $buffer = fread($this->socket, $need);
            
            if ($buffer === false) {
                $this->logError("读取失败，已读取" . strlen($data) . "字节");
                $this->connected = false;
                return false;
            }
            
            if ($buffer === '') {
                // 检查是否超时
                if (time() - $startTime > 60) {
                    $this->logError("读取超时，已读取" . strlen($data) . "字节");
                    $this->connected = false;
                    return false;
                }
                usleep(100000); // 等待100ms再试
                continue;
            }
            
            $data .= $buffer;
        }
        return $data;
    }

    /**
     * 关闭连接
     */
    public function close() {
        if ($this->socket) {
            try {
                // 发送关闭帧
                fwrite($this->socket, chr(0x88) . chr(0x00));
                fclose($this->socket);
            } catch (\Exception $e) {
                $this->logError("关闭连接失败: " . $e->getMessage());
            }
            $this->socket = null;
        }
        $this->connected = false;
    }

    /**
     * 检查是否已连接
     * @return bool
     */
    public function isConnected() {
        return $this->connected && $this->socket !== null;
    }

    /**
     * 编码WebSocket数据
     * @param string $data
     * @return string
     */
    private function encode($data) {
        $length = strlen($data);
        $header = chr(0x81); // FIN=1, opcode=1(文本)
        $mask = chr(rand(0, 255)) . chr(rand(0, 255)) . chr(rand(0, 255)) . chr(rand(0, 255));

        // 处理长度
        if ($length <= 125) {
            $header .= chr(0x80 | $length);
        } elseif ($length < 65536) {
            $header .= chr(0x80 | 126) . pack('n', $length);
        } else {
            $header .= chr(0x80 | 127) . pack('J', $length);
        }

        // 应用掩码
        $payload = '';
        for ($i = 0; $i < $length; $i++) {
            $payload .= $data[$i] ^ $mask[$i % 4];
        }

        return $header . $mask . $payload;
    }

    /**
     * 日志记录（错误）
     * @param string $message
     */
    private function logError($message) {
        $this->log("[ERROR] WebSocket: {$message}");
    }

    /**
     * 日志记录（信息）
     * @param string $message
     */
    private function logInfo($message) {
        $this->log("[INFO] WebSocket: {$message}");
    }

    /**
     * 通用日志方法
     */
    private function log($message) {
        $logFile = ROOT_D . '/websocket_logs.txt';
        if (!file_exists(dirname($logFile))) {
            mkdir(dirname($logFile), 0755, true);
        }
        file_put_contents($logFile, date('[Y-m-d H:i:s] ') . $message . "\n", FILE_APPEND);
    }
}
?>